package com.ljh.seckill.mq;

import com.ljh.seckill.config.SeckillConfig;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

/**
 * @Author grung
 * @Date 2021/2/23 18:39
 * @Version 1.0
 */

@Component
public class Consumer implements ApplicationListener<ContextRefreshedEvent> {
    private static Logger logger = LoggerFactory.getLogger(Consumer.class);

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;
    @Value("${apache.rocketmq.consumer.groupName}")
    private String groupName;

    @Autowired
    private ConsumerListener consumerListener;

    volatile boolean isStart = false;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {

        if (isStart) {
            return;
        }
        isStart = true;
        logger.info("启动RocketMQ消费者监听器....CONSUMER_GROUP:{}", groupName);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.registerMessageListener(consumerListener);
        consumer.setVipChannelEnabled(false);
        consumer.setConsumerGroup(groupName);
        /**
         * Minimum consumer thread number 默认20
         */
        consumer.setConsumeThreadMin(1);
        /**
         * Max consumer thread number 默认20
         */
        consumer.setConsumeThreadMax(1);
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         * 当前采用队列头消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 设置一次消费消息的条数，默认为1条
         */
        consumer.setConsumeMessageBatchMaxSize(1);
        /**
         * Max re-consume times. -1 means 16 times.
         */
        consumer.setMaxReconsumeTimes(5);

        try {
            /**
             * 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，则tag使用*；如果需要指定订阅该主题下的某些tag，则使用||分割，例如tag1||tag2||tag3
             */
            consumer.subscribe(SeckillConfig.ROCKETMQ_TOPIC_ORDER, "*");
            consumer.start();
            logger.info("RocketMQ消费者监听器启动成功");
        } catch (Exception e) {
            logger.error("RocketMQ消费者监听器启动失败", e);
        }
    }
}
